MCP 客户端的工程实践与提示词设计
办公场景需求分析
在现代办公环境中，员工经常需要处理各种重复性任务：文档管理、邮件处理、日程安排、数据分析等。通过 MCP 客户端构建智能办公助手，可以显著提升工作效率。
核心功能模块设计
核心架构实现
主控制器设计
import asyncio
import json
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
class TaskType(Enum):
DOCUMENT = "document"
EMAIL = "email"
CALENDAR = "calendar"
DATA_ANALYSIS = "data_analysis"
KNOWLEDGE = "knowledge"
@dataclass
class OfficeTask:
task_type: TaskType
description: str
parameters: Dict[str, Any]
priority: int = 1 # 1-5, 5最高
context: Optional[str] = None
class OfficeAssistantClient:
def __init__(self, ai_client, mcp_sessions: Dict[str, Any]):
self.ai_client = ai_client
self.mcp_sessions = mcp_sessions # 各个MCP服务的连接
self.available_tools = {}
self.conversation_history = []
self.current_context = {}
async def initialize(self):
"""初始化所有MCP连接并注册工具"""
for service_name, session in self.mcp_sessions.items():
await session.initialize()
tools = await self._register_service_tools(service_name, session)
self.available_tools[service_name] = tools
logging.info(f"已注册 {len(self.available_tools)} 个服务模块")
智能工具选择系统
class ToolSelector:
def __init__(self):
self.selection_history = []
def get_system_prompt(self, available_tools: List[Dict]) -> str:
"""生成工具选择的系统提示词"""
tools_description = self._format_tools_for_prompt(available_tools)
return f"""你是一个专业的办公助手，能够理解用户的工作需求并选择最合适的工具来完成任务。
可用工具列表：
{tools_description}
工具选择原则：
1. **精确匹配**：优先选择功能最匹配的工具
2. **效率优先**：选择能最快完成任务的工具组合
3. **上下文感知**：考虑当前工作场景和历史操作
4. **错误处理**：如果工具执行失败，主动选择备选方案
响应格式要求：
- 如果需要调用工具，使用标准的tool_calls格式
- 如果需要多个步骤，先调用第一个工具，后续根据结果继续
- 如果无法完成任务，明确说明原因和建议
当前工作场景：办公环境，注重效率和准确性。"""
def _format_tools_for_prompt(self, tools: List[Dict]) -> str:
"""格式化工具描述 供AI理解"""
formatted = []
for tool in tools:
func = tool['function']
params_desc = self._format_parameters(func.get('parameters', {}))
formatted.append(f"""
**{func['name']}**
- 功能：{func['description']}
- 参数：{params_desc}
- 适用场景：{self._get_tool_scenarios(func['name'])}
""")
return "\n".join(formatted)
def _get_tool_scenarios(self, tool_name: str) -> str:
"""根据工具名称返回适用场景描述"""
scenarios = {
'create_document': '创建新文档、草拟报告、制作模板',
'send_email': '发送通知、回复邮件、群发消息',
'schedule_meeting': '安排会议、设置提醒、协调时间',
'analyze_data': '数据分析、生成图表、统计报告',
'search_knowledge': '查找资料、技术文档、政策查询'
}
return scenarios.get(tool_name, '通用办公任务')
上下文感知的任务处理
class ContextAwareProcessor:
def __init__(self, client):
self.client = client
self.context_weights = {
'time_context': 0.3, # 时间相关性
'topic_context': 0.4, # 主题相关性
'user_preference': 0.3 # 用户偏好
}
async def process_user_input(self, user_input: str) -> str:
"""处理用户输入，考虑上下文信息"""
# 1. 构建富上下文的prompt
context_prompt = self._build_context_prompt(user_input)
# 2. 第一轮AI交互：理解意图和选择工具
messages = [
{"role": "system", "content": context_prompt},
{"role": "user", "content": user_input}
]
response = await self.client.ai_client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=self._get_available_tools(),
temperature=0.1 # 降低随机性，提高工具选择准确性
)
# 3. 处理工具调用
if response.choices[0].message.tool_calls:
return await self._execute_tool_workflow(response, messages)
else:
return response.choices[0].message.content
def _build_context_prompt(self, user_input: str) -> str:
"""构建包含上下文信息的提示词"""
current_time = datetime.now().strftime("%Y-%m-%d %H:%M")
recent_tasks = self._get_recent_tasks()
user_preferences = self._get_user_preferences()
return f"""你是一个智能办公助手，当前时间：{current_time}
用户最近的工作任务：
{recent_tasks}
用户工作偏好：
{user_preferences}
请基于以上上下文信息，理解用户需求并选择合适的工具。注意：
1. 考虑时间敏感性（紧急任务优先）
2. 关注工作连续性（相关任务串联）
3. 尊重用户习惯（常用工具优先）
{self.client.tool_selector.get_system_prompt(self._get_available_tools())}"""
具体场景实现
智能邮件处理
class EmailProcessor:
def __init__(self, mcp_client):
self.mcp_client = mcp_client
async def process_email_request(self, user_input: str, context: Dict):
"""处理邮件相关请求"""
# 邮件处理专用提示词
email_prompt = """
你是邮件处理专家，可以帮助用户：
1. 撰写正式邮件（商务、通知、邀请等）
2. 回复邮件（根据收到的邮件内容生成回复）
3. 邮件分类和整理
4. 批量邮件操作
邮件撰写要求：
- 商务邮件：正式、简洁、重点突出
- 内部邮件：友好、直接、高效
- 客户邮件：专业、礼貌、解决问题导向
请根据用户需求，选择合适的邮件操作工具。
"""
# 检测邮件类型和紧急程度
email_analysis = await self._analyze_email_intent(user_input)
if email_analysis['type'] == 'compose':
return await self._handle_email_composition(user_input, email_analysis)
elif email_analysis['type'] == 'reply':
return await self._handle_email_reply(user_input, email_analysis)
else:
return await self._handle_email_management(user_input, email_analysis)
async def _handle_email_composition(self, user_input: str, analysis: Dict):
"""处理邮件撰写"""
# 构建邮件撰写的详细prompt
compose_prompt = f"""
请帮助撰写邮件，要求：
邮件类型：{analysis.get('business_type', '商务邮件')}
收件人：{analysis.get('recipients', '未指定')}
紧急程度：{analysis.get('urgency', '普通')}
语言风格：{analysis.get('tone', '正式')}
邮件结构要求：
1. 主题行：简洁明了，突出重点
2. 开头：适当的称呼和问候
3. 正文：逻辑清晰，重点突出
4. 结尾：明确的行动项或期望
5. 签名：标准商务签名
用户需求：{user_input}
"""
# 调用邮件撰写工具
result = await self.mcp_client.call_tool('compose_email', {
'prompt': compose_prompt,
'template_type': analysis.get('template_type', 'business'),
'urgency': analysis.get('urgency', 'normal')
})
return result
文档智能管理
class DocumentManager:
def __init__(self, mcp_client):
self.mcp_client = mcp_client
async def handle_document_task(self, user_input: str):
"""处理文档相关任务"""
# 文档任务分类prompt
doc_classification_prompt = """
分析用户的文档需求，判断任务类型：
1. CREATE：创建新文档（报告、方案、总结等）
2. EDIT：编辑现有文档（修改、更新、格式调整）
3. CONVERT：格式转换（PDF转Word、Excel转CSV等）
4. SEARCH：文档搜索（查找特定文档或内容）
5. ANALYZE：文档分析（提取关键信息、统计等）
请识别任务类型并提取关键参数。
"""
task_type = await self._classify_document_task(user_input)
if task_type == 'CREATE':
return await self._create_document(user_input)
elif task_type == 'SEARCH':
return await self._search_documents(user_input)
# ... 其他任务类型处理
async def _create_document(self, user_input: str):
"""智能文档创建"""
# 文档创建的结构化prompt
creation_prompt = f"""
请帮助创建文档，要求：
内容分析：
- 从用户需求中提取文档类型、主题、关键要素
- 确定文档结构和章节安排
- 识别所需数据和信息来源
文档规范：
- 使用标准商务文档格式
- 包含目录、正文、结论
- 保持逻辑清晰、数据准确
用户需求：{user_input}
请生成文档大纲，然后调用相应工具创建文档。
"""
# 多步骤文档创建流程
outline = await self._generate_document_outline(creation_prompt)
content = await self._generate_document_content(outline)
result = await self.mcp_client.call_tool('create_document', {
'title': outline['title'],
'content': content,
'format': outline.get('format', 'docx'),
'template': outline.get('template', 'business_report')
})
return result
数据分析助手
class DataAnalysisAssistant:
def __init__(self, mcp_client):
self.mcp_client = mcp_client
async def analyze_data_request(self, user_input: str, data_context: Dict):
"""处理数据分析请求"""
analysis_prompt = f"""
你是数据分析专家，能够：
1. 理解业务需求并转换为分析任务
2. 选择合适的分析方法和可视化方式
3. 生成专业的分析报告
当前数据上下文：
- 数据源：{data_context.get('data_sources', '未知')}
- 数据类型：{data_context.get('data_types', '未知')}
- 时间范围：{data_context.get('time_range', '未知')}
分析要求：
- 提供清晰的数据洞察
- 使用适当的图表类型
- 突出关键指标和趋势
- 提供可行的业务建议
用户需求：{user_input}
"""
# 分析类型识别
analysis_type = await self._identify_analysis_type(user_input)
if analysis_type == 'trend_analysis':
return await self._perform_trend_analysis(user_input, data_context)
elif analysis_type == 'comparison':
return await self._perform_comparison_analysis(user_input, data_context)
# ... 其他分析类型
async def _perform_trend_analysis(self, user_input: str, context: Dict):
"""执行趋势分析"""
result = await self.mcp_client.call_tool('analyze_trends', {
'data_source': context.get('data_source'),
'metrics': context.get('metrics', []),
'time_period': context.get('time_range'),
'visualization_type': 'line_chart',
'analysis_prompt': f"分析以下趋势并提供业务洞察：{user_input}"
})
return result
高级提示词策略
多轮对话管理
class ConversationManager:
def __init__(self):
self.conversation_state = {}
self.context_window = 10 # 保持最近10轮对话
def build_conversation_prompt(self, current_input: str, user_id: str) -> List[Dict]:
"""构建考虑对话历史的prompt"""
history = self.get_conversation_history(user_id)
# 系统prompt包含对话状态
system_prompt = f"""
你是智能办公助手，正在与用户进行多轮对话。
对话状态：
- 当前任务：{self.conversation_state.get(user_id, {}).get('current_task', '无')}
- 上下文信息：{self.conversation_state.get(user_id, {}).get('context', {})}
- 待完成步骤：{self.conversation_state.get(user_id, {}).get('pending_steps', [])}
请基于对话历史和当前状态，提供连贯、有用的回复。
如果是延续之前的任务，请明确引用相关上下文。
"""
messages = [{"role": "system", "content": system_prompt}]
# 添加对话历史（保持在token限制内）
for msg in history[-self.context_window:]:
messages.append(msg)
messages.append({"role": "user", "content": current_input})
return messages
错误恢复机制
class ErrorRecoveryHandler:
def __init__(self, client):
self.client = client
self.retry_strategies = {
'tool_call_failed': self._handle_tool_failure,
'parsing_error': self._handle_parsing_error,
'timeout': self._handle_timeout
}
async def handle_error(self, error_type: str, context: Dict, original_request: str):
"""智能错误恢复"""
recovery_prompt = f"""
检测到执行错误：{error_type}
错误上下文：{context}
原始请求：{original_request}
请分析错误原因并提供恢复方案：
1. 如果是工具参数问题，调整参数重试
2. 如果是工具不可用，选择替代方案
3. 如果是网络问题，建议稍 后重试
4. 如果无法恢复，明确告知用户并提供手动操作建议
恢复策略应该：
- 保持用户任务的连续性
- 提供清晰的错误说明
- 给出具体的解决方案
"""
recovery_plan = await self.client.ai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": recovery_prompt}],
temperature=0.1
)
return await self._execute_recovery_plan(recovery_plan, context)
性能优化实现
class PerformanceOptimizer:
def __init__(self):
self.cache = {}
self.tool_performance = {}
async def optimize_tool_selection(self, user_intent: str, available_tools: List):
"""基于性能数据优化工具选择"""
# 计算工具性能评分
tool_scores = {}
for tool in available_tools:
tool_name = tool['function']['name']
performance_data = self.tool_performance.get(tool_name, {})
# 综合评分：成功率 * 速度 * 用户满意度
score = (
performance_data.get('success_rate', 0.5) * 0.4 +
(1 - performance_data.get('avg_response_time', 5) / 10) * 0.3 +
performance_data.get('user_satisfaction', 0.5) * 0.3
)
tool_scores[tool_name] = score
# 在prompt中包含性能建议
performance_prompt = f"""
工具性能建议（基于历史数据）：
{self._format_performance_suggestions(tool_scores)}
请优先选择高性能工具，除非特殊需求要求使用特定工具。
"""
return performance_prompt
def _format_performance_suggestions(self, scores: Dict) -> str:
"""格式化性能建议"""
sorted_tools = sorted(scores.items(), key=lambda x: x[1], reverse=True)
suggestions = []
for tool, score in sorted_tools[:3]: # 推荐前3个
rating = "★★★★★" if score > 0.8 else "★★★☆☆" if score > 0.6 else "★★☆☆☆"
suggestions.append(f"- {tool}: {rating} (推荐度: {score:.2f})")
return "\n".join(suggestions)
部署和监控
配置管理
# config.py
import os
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class OfficeAssistantConfig:
# AI模型配置
ai_model: str = "gpt-4"
ai_temperature: float = 0.1
max_tokens: int = 4000
# MCP服务配置
mcp_services: Dict[str, str] = None
# 性能配置
max_concurrent_tools: int = 3
tool_timeout: int = 30
cache_ttl: int = 3600
# 日志配置
log_level: str = "INFO"
log_file: str = "office_assistant.log"
def __post_init__(self):
if self.mcp_services is None:
self.mcp_services = {
'document_service': os.getenv('DOCUMENT_MCP_URL', 'stdio:document-server'),
'email_service': os.getenv('EMAIL_MCP_URL', 'stdio:email-server'),
'calendar_service': os.getenv('CALENDAR_MCP_URL', 'stdio:calendar-server'),
'data_service': os.getenv('DATA_MCP_URL', 'stdio:data-server')
}
# 环境变量配置示例
config = OfficeAssistantConfig(
ai_model=os.getenv('AI_MODEL', 'gpt-4'),
max_concurrent_tools=int(os.getenv('MAX_CONCURRENT_TOOLS', '3')),
)
监控和日志
import asyncio
import time
from datetime import datetime
import json
class OfficeAssistantMonitor:
def __init__(self):
self.metrics = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'tool_calls': {},
'response_times': [],
'user_satisfaction_scores': []
}
async def log_request(self, user_input: str, response: str,
tools_used: List[str], response_time: float):
"""记录请求日志"""
self.metrics['total_requests'] += 1
self.metrics['response_times'].append(response_time)
for tool in tools_used:
self.metrics['tool_calls'][tool] = self.metrics['tool_calls'].get(tool, 0) + 1
log_entry = {
'timestamp': datetime.now().isoformat(),
'user_input': user_input[:100], # 截断长输入
'response_length': len(response),
'tools_used': tools_used,
'response_time': response_time,
'success': True # 根据实际情况判断
}
logging.info(f"Request processed: {json.dumps(log_entry)}")
def get_performance_report(self) -> Dict:
"""生成性能报告"""
if not self.metrics['response_times']:
return {'status': 'No data available'}
avg_response_time = sum(self.metrics['response_times']) / len(self.metrics['response_times'])
success_rate = self.metrics['successful_requests'] / self.metrics['total_requests']
return {
'total_requests': self.metrics['total_requests'],
'success_rate': f"{success_rate:.2%}",
'avg_response_time': f"{avg_response_time:.2f}s",
'most_used_tools': sorted(
self.metrics['tool_calls'].items(),
key=lambda x: x[1],
reverse=True
)[:5]
}
总结
通过精心设计的提示词策略和工程化实现，办公场景的 MCP 客户端能够：
- 智能理解：准确识别用户意图和任务类型
- 工具选择：基于上下文和性能数据选择最优工具
- 错误恢复：提供robust的错误处理和恢复机制
- 性能优化：通过缓存、并发控制等手段提升响应速度
- 可观测性：完整的日志和监控体系
关键成功要素：
- 提示词工程：针对不同场景设计专门的提示词模板
- 上下文管理：维护对话状态和工作上下文
- 性能监控：持续优化工具选择和执行效率
- 用户体验：提供连贯、智能的交互体验
这样的设计能够真正实现"AI赋能办公"的目标，让AI助手成为用户高效工作的得力伙伴。